package channel

import (
	"runtime"
	"sync/atomic"
	"time"
)

// PriorityChannel 具有优先级的双通道
type PriorityChannel interface {
	Start()                                // 开启任务
	Stop()                                 // 结束任务
	DispatchPriorTask(task any)            // 指派高优先级任务
	DispatchNormalTask(task any)           // 指派正常任务
	SetPriorWorker(worker func(task any))  // 设置任务执行器
	SetNormalWorker(worker func(task any)) // 设置任务执行器
	NormalTaskIsFull() bool                // 采集任务队列是否已满
	GetMetrics() ChannelMetrics            // 获取通道指标
}

// ChannelMetrics 通道性能指标
type ChannelMetrics struct {
	PriorQueueSize    int   `json:"prior_queue_size"`
	NormalQueueSize   int   `json:"normal_queue_size"`
	PriorTasksTotal   int64 `json:"prior_tasks_total"`
	NormalTasksTotal  int64 `json:"normal_tasks_total"`
	DroppedTasksTotal int64 `json:"dropped_tasks_total"`
	WorkerActive      bool  `json:"worker_active"`
}

var _ PriorityChannel = (*priorityChannel)(nil)

type priorityChannel struct {
	// 通道配置 - 根据系统资源动态调整
	priorChan  chan any
	normalChan chan any
	stopChan   chan struct{}

	// 工作器函数
	priorWorker  func(any)
	normalWorker func(any)

	// 性能指标 (原子操作)
	priorTasksTotal   int64
	normalTasksTotal  int64
	droppedTasksTotal int64
	workerActive      int32 // 0=inactive, 1=active

	// 配置参数
	maxPriorSize   int
	maxNormalSize  int
	maxNormalCheck int
}

// NewPriorityChannel 创建优先级通道
func NewPriorityChannel() PriorityChannel {
	// 根据CPU核心数动态调整通道容量
	cpuCount := runtime.NumCPU()

	// 优先级通道容量：CPU核心数 * 64
	priorSize := cpuCount * 64
	if priorSize < 128 {
		priorSize = 128
	} else if priorSize > 2048 {
		priorSize = 2048
	}

	// 普通通道容量：CPU核心数 * 1024
	normalSize := cpuCount * 1024
	if normalSize < 1024 {
		normalSize = 1024
	} else if normalSize > 16384 {
		normalSize = 16384
	}

	return &priorityChannel{
		priorChan:      make(chan any, priorSize),
		normalChan:     make(chan any, normalSize),
		stopChan:       make(chan struct{}, 1),
		maxPriorSize:   priorSize,
		maxNormalSize:  normalSize,
		maxNormalCheck: normalSize * 4 / 5, // 80% 阈值
	}
}

func (pc *priorityChannel) Start() {
	atomic.StoreInt32(&pc.workerActive, 1)
	go pc.Worker()
}

func (pc *priorityChannel) Stop() {
	atomic.StoreInt32(&pc.workerActive, 0)
	select {
	case pc.stopChan <- struct{}{}:
	default:
		// 通道已满，忽略
	}
}

func (pc *priorityChannel) DispatchPriorTask(task any) {
	select {
	case pc.priorChan <- task:
		atomic.AddInt64(&pc.priorTasksTotal, 1)
	default:
		// 优先级通道满了，丢弃任务（这种情况很少见）
		atomic.AddInt64(&pc.droppedTasksTotal, 1)
	}
}

func (pc *priorityChannel) DispatchNormalTask(task any) {
	select {
	case pc.normalChan <- task:
		atomic.AddInt64(&pc.normalTasksTotal, 1)
	default:
		// 普通通道满了，丢弃任务
		atomic.AddInt64(&pc.droppedTasksTotal, 1)
	}
}

func (pc *priorityChannel) SetPriorWorker(worker func(task any)) {
	pc.priorWorker = worker
}

func (pc *priorityChannel) SetNormalWorker(worker func(task any)) {
	pc.normalWorker = worker
}

func (pc *priorityChannel) NormalTaskIsFull() bool {
	return len(pc.normalChan) >= pc.maxNormalCheck
}

func (pc *priorityChannel) GetMetrics() ChannelMetrics {
	return ChannelMetrics{
		PriorQueueSize:    len(pc.priorChan),
		NormalQueueSize:   len(pc.normalChan),
		PriorTasksTotal:   atomic.LoadInt64(&pc.priorTasksTotal),
		NormalTasksTotal:  atomic.LoadInt64(&pc.normalTasksTotal),
		DroppedTasksTotal: atomic.LoadInt64(&pc.droppedTasksTotal),
		WorkerActive:      atomic.LoadInt32(&pc.workerActive) == 1,
	}
}

// Worker 优化的优先级处理算法
// 使用时间片和饥饿防护机制
func (pc *priorityChannel) Worker() {
	defer atomic.StoreInt32(&pc.workerActive, 0)

	normalTasksBatch := 0
	const maxNormalBatch = 5 // 连续处理5个普通任务后，必须检查优先级任务

	ticker := time.NewTicker(100 * time.Millisecond) // 100ms检查一次状态
	defer ticker.Stop()

	for {
		select {
		case <-pc.stopChan:
			return

		case <-ticker.C:
			// 定期检查，防止goroutine泄漏
			if atomic.LoadInt32(&pc.workerActive) == 0 {
				return
			}
			continue

		case priorTask := <-pc.priorChan:
			// 处理优先级任务
			if pc.priorWorker != nil {
				pc.priorWorker(priorTask)
			}
			normalTasksBatch = 0 // 重置普通任务批次计数

		case normalTask := <-pc.normalChan:
			// 检查是否需要让出给优先级任务
			if normalTasksBatch >= maxNormalBatch {
				// 快速检查是否有优先级任务
				select {
				case priorTask := <-pc.priorChan:
					if pc.priorWorker != nil {
						pc.priorWorker(priorTask)
					}
					normalTasksBatch = 0
					continue
				default:
					// 没有优先级任务，继续处理普通任务
				}
			}

			// 处理普通任务
			if pc.normalWorker != nil {
				pc.normalWorker(normalTask)
			}
			normalTasksBatch++
		}
	}
}

// WorkerV1 原始算法保留（备用）
// 算法参见：https://blog.csdn.net/hurray123/article/details/50038329/
func (pc *priorityChannel) WorkerV1() {
	defer atomic.StoreInt32(&pc.workerActive, 0)

	for {
		select {
		case <-pc.stopChan:
			return
		case priorTask := <-pc.priorChan:
			if pc.priorWorker != nil {
				pc.priorWorker(priorTask)
			}
		default:
			select {
			case priorTask := <-pc.priorChan:
				if pc.priorWorker != nil {
					pc.priorWorker(priorTask)
				}
			case normalTask := <-pc.normalChan:
				if pc.normalWorker != nil {
					pc.normalWorker(normalTask)
				}
			}
		}
	}
}

// WorkerV2 另外一种写法保留（备用）
// https://www.csdn.net/tags/MtTaQg4sMTk5LWJsb2cO0O0O.html
// https://www.zhihu.com/question/469460715/answer/1974574115
func (pc *priorityChannel) WorkerV2() {
	defer atomic.StoreInt32(&pc.workerActive, 0)

	for {
		select {
		case <-pc.stopChan:
			return
		case priorTask := <-pc.priorChan:
			if pc.priorWorker != nil {
				pc.priorWorker(priorTask)
			}
		case normalTask := <-pc.normalChan:
		priority:
			for {
				select {
				case priorTask := <-pc.priorChan:
					if pc.priorWorker != nil {
						pc.priorWorker(priorTask)
					}
				default:
					break priority
				}
			}
			if pc.normalWorker != nil {
				pc.normalWorker(normalTask)
			}
		}
	}
}
